package com.mio.kududemo;

import java.util.LinkedList;

import org.apache.kudu.Schema;
import org.apache.kudu.client.CreateTableOptions;
import org.apache.kudu.client.Delete;
import org.apache.kudu.client.Insert;
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.client.KuduException;
import org.apache.kudu.client.KuduScanner;
import org.apache.kudu.client.KuduSession;
import org.apache.kudu.client.KuduTable;
import org.apache.kudu.client.PartialRow;
import org.apache.kudu.client.RowResult;
import org.apache.kudu.client.SessionConfiguration;
import org.apache.kudu.client.Update;

public class KuduUtil {

	public static void createTable(KuduClient client, Schema schema, String tableName, String partitionKey) {
		// 创建表时提供的所有选项
		CreateTableOptions options = new CreateTableOptions();
		// 设置表的replica备份和分区规则
		LinkedList<String> parcols = new LinkedList<>();
		parcols.add(partitionKey);
		// 设置表的备份数
		options.setNumReplicas(1);
		// 设置hash分区和数量
		options.addHashPartitions(parcols, 3);
		try {
			client.createTable(tableName, schema, options);
		} catch (KuduException e) {
			e.printStackTrace();
		}

	}
	public static void search(KuduClient client,String table) throws KuduException {
		 KuduTable stuTable = client.openTable(table);
         KuduScanner scanner = client.newScannerBuilder(stuTable).build();
         while(scanner.hasMoreRows()) {
             for(RowResult result :scanner.nextRows()) {
                 int gid = result.getInt("id");
                 String gname = result.getString("name");
                 System.out.println("id:" + gid + "...name:" + gname);
             }

         }
	}
	
	public static void insert(KuduClient client,String table, int id,String name) throws KuduException {
		 //2、打开一张表
        KuduTable stuTable = client.openTable(table);
        //3、创建会话
        KuduSession kuduSession = client.newSession();
        //4、设置刷数据模式
        kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
        //5、获取插入实例
        Insert insert = stuTable.newInsert();
        //6、声明待插入数据
        insert.getRow().addInt("id",id);
        insert.getRow().addString("name",name);
        //7、刷入数据
        kuduSession.flush();
        //8、应用插入实例
        kuduSession.apply(insert);
        //9、关闭会话
        kuduSession.close();
	}
	public static void update(KuduClient client,String table, int id,String name) throws KuduException {
		  KuduTable stuTable = client.openTable(table);
          KuduSession kuduSession = client.newSession();
          kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
          Update update = stuTable.newUpdate();
          PartialRow row = update.getRow();
          row.addInt("id",id);
          row.addString("name",name);
          kuduSession.apply(update);

          kuduSession.close();
	}
	
	public static void delete(KuduClient client,String table, int id) throws KuduException {
		  KuduTable stuTable = client.openTable(table);
          KuduSession kuduSession = client.newSession();
          kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);

          Delete delete = stuTable.newDelete();
          PartialRow row = delete.getRow();
          row.addInt("id",id);

          kuduSession.flush();
          kuduSession.apply(delete);

          kuduSession.close();
	}
}
